package net.pimpas.communication.channel.impl;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

import net.pimpas.communication.channel.Channel;
import net.pimpas.communication.exception.CommunicationException;
import net.pimpas.communication.exception.NotifyException;
import net.pimpas.communication.listener.ChannelListener;
import net.pimpas.communication.message.Message;
import net.pimpas.core.configuration.exception.ConfigurationException;
import net.pimpas.executor.ExecutorThreadType;
import net.pimpas.executor.serviceprovider.ExecutorServiceProvider;
import net.pimpas.executor.serviceprovider.impl.ExecutorServiceProviderFactoryImpl;
import net.pimpas.logging.serviceprovider.LoggingServiceProvider;
import net.pimpas.logging.serviceprovider.LoggingServiceProviderFactoryImpl;

public class DefaultChannelImpl implements Channel {

	private ExecutorServiceProvider executor;
	private LoggingServiceProvider logger = 
		LoggingServiceProviderFactoryImpl.getLoggingServiceProvider(DefaultChannelImpl.class);

	private boolean closed;
	private Date createdTime;
	private Date lastMessageTime;
	private String id;
	private List<ChannelListener> listeners;
	private List<Message> messages;

	public DefaultChannelImpl(String id) {
		this(id, null);
	}

	public DefaultChannelImpl(String id, ExecutorServiceProvider executor) {
		this.setClosed(false);
		this.setId(id);
		this.setListeners(new ArrayList<ChannelListener>());
		this.setMessages(new ArrayList<Message>());
		try {
			if(executor == null) {
				this.setExecutor(ExecutorServiceProviderFactoryImpl.getExecutorServiceProvider("Channel_"+this.getId()));
				this.getExecutor().configure(ExecutorThreadType.CACHED_THREAD_POOL);
			} else
				this.setExecutor(executor);
		} catch (ConfigurationException e) {
			throw new IllegalStateException(e);
		}
	}

	public void close() {
		this.checkClosed();
		this.setClosed(true);
		this.getExecutor().shutdown();
		try {
			this.getExecutor().awaitTermination(1, TimeUnit.MINUTES);
		} catch (InterruptedException e) {
		} finally {
			synchronized(this.getMessages()) {
				this.getMessages().clear();
			}
			synchronized(this.getListeners()) {
				this.getListeners().clear();
			}
			
			ExecutorServiceProviderFactoryImpl.terminateExecutor(this.getExecutor());
		}
	}

	public Date getCreatedTime() {
		return this.createdTime;
	}

	public String getId() {
		return this.id;
	}

	public Date getLastMessageTime() {
		return this.lastMessageTime;
	}

	public boolean isTerminated() {
		return this.isClosed();
	}
	
	private void checkClosed( ) {
		if(this.isClosed()) throw new IllegalStateException("channel already closed!");
	}

	public void subscribe(ChannelListener listener) {
		this.checkClosed();
		
		if(this.getListeners().contains(listener)) return;
		this.getLogger().debug("[Channel]: Listener ["+listener+"] subscribe channel with id ("+this.getId()+")]");

		synchronized(getListeners()) {
			this.getListeners().add(listener);
		}
	}

	public List<Message> unsubscribe(ChannelListener listener) {
		this.checkClosed();
		
		if(!this.getListeners().contains(listener)) return null;

		synchronized(getListeners()) {
			this.getListeners().remove(listener);
		}

		// get all listener "non-read" messages and return
		return null;
	}

	public boolean isClosed() {
		return closed;
	}

	public void setClosed(boolean closed) {
		this.closed = closed;
	}

	public ExecutorServiceProvider getExecutor() {
		return executor;
	}

	public void setExecutor(ExecutorServiceProvider executor) {
		this.executor = executor;
	}

	public LoggingServiceProvider getLogger() {
		return logger;
	}

	public void setLogger(LoggingServiceProvider logger) {
		this.logger = logger;
	}

	public void setCreatedTime(Date createdTime) {
		this.createdTime = createdTime;
	}

	public void setLastMessageTime(Date lastMessageTime) {
		this.lastMessageTime = lastMessageTime;
	}

	public void setId(String id) {
		this.id = id;
	}

	public List<ChannelListener> getListeners() {
		return listeners;
	}

	public void setListeners(List<ChannelListener> listeners) {
		this.listeners = listeners;
	}

	public List<Message> getMessages() {
		return messages;
	}

	public void setMessages(List<Message> messages) {
		this.messages = messages;
	}

	public void send(final Message message) throws CommunicationException {
		this.checkClosed();
		
		assert(message!=null);
		
		synchronized(this) {
			this.setLastMessageTime(Calendar.getInstance().getTime());
		}
		
		this.getExecutor().execute(new ExecutorRunnable(this, message, this.getListeners()));
		
		/*
		synchronized(getMessages()) {
			this.getMessages().add(message);
		}

		this.getExecutor().execute(new Runnable() {
			public void run() {
				checkClosed();
				logger.debug("[Channel]: Broadcast message for all listeners.");
				logger.debug("[Channel]: Message: "+message);
				final List<ChannelListener> ls = getListeners();
				try {
					for(ChannelListener listener: ls) {
						checkClosed();
						try {
							listener.notify(DefaultChannelImpl.this, message);
						} catch (NotifyException e) {
							logger.error("[Channel]: Error in notify channellistener!", e);
						}
					}
				} finally {
					synchronized(getMessages()) {
						getMessages().remove(message);
					}
				}
			}
		});
		*/
	}
	
	private final static class ExecutorRunnable implements Runnable {
		
		private Message message;
		private Channel channel;
		private List<ChannelListener> listeners;
		private static final LoggingServiceProvider logger = 
			LoggingServiceProviderFactoryImpl.getLoggingServiceProvider(ExecutorRunnable.class);
		
		public ExecutorRunnable(Channel channel, Message message, List<ChannelListener> listeners) {
			this.setMessage(message);
			this.setChannel(channel);
			this.setListeners(listeners);
		}
		
		private void checkClosed( ) {
			if(channel.isTerminated()) {
				logger.debug("[Channel "+this.getChannel()+" is closed!]");
				throw new IllegalStateException("channel already closed!");
			}
		}

		public void run() {
			logger.debug("[Channel]: Broadcast message for all listeners.");
			logger.debug("[Channel]: Message: "+message);
			for(ChannelListener listener: this.getListeners()) {
				checkClosed();
				try {
					listener.notify(this.getChannel(), this.getMessage());
				} catch (NotifyException e) {
					logger.error("[Channel]: Error in notify channellistener!", e);
				}
			}
		}

		public List<ChannelListener> getListeners() {
			return listeners;
		}

		public void setListeners(List<ChannelListener> listeners) {
			this.listeners = listeners;
		}

		public Message getMessage() {
			return message;
		}

		public void setMessage(Message message) {
			this.message = message;
		}

		public Channel getChannel() {
			return channel;
		}

		public void setChannel(Channel channel) {
			this.channel = channel;
		}
	}
}